﻿//
// File: QueuedOperation.cs
//
// Purpose:  implements queued execution classes,
//           sequenced and and paraller executor. 
//           quite rudimentary and functional.
//
// Author: MotherHacker @ cachecleaner.codeplex.com


using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

using System.Runtime.InteropServices;
using System.Security.Permissions;
using Win32Exception = System.ComponentModel.Win32Exception;

namespace QueuedOperation
{
    #region Interfaces
    /// <summary>
    /// Implements executor.
    /// </summary>
    public interface IExecutor
    {
        void Execute( IRunnable exe );
    }

    /// <summary>
    /// Implements runnable.
    /// </summary>
    public interface IRunnable
    {
        void Run();
    }

    /// <summary>
    /// Implements operation.
    /// </summary>
    public interface IOperation : IRunnable
    {
        /// <summary>
        /// Determines if the operation has completed.
        /// </summary>
        bool IsDone
        {
            get;
        }

        /// <summary>
        /// If exception occurs, this will be set.
        /// </summary>
        Exception ThrownException
        {
            get;
            set;
        }

        /// <summary>
        /// Wait till operation has completed.
        /// </summary>
        bool WaitDone();

        /// <summary>
        /// Wait operation to completed with given time-out.
        /// </summary>
        /// <param name="timeOut">time out in milliseconds.</param>
        /// <returns>true if operation completed in given time out, otherwise false.</returns>
        bool WaitDone(int timeOut);
    }
    #endregion

    #region AsyncOperation
    /// <summary>
    /// Implements async operation, that will be executed later in time.
    /// </summary>
    public abstract class AsyncOperation : IOperation
    {
        ManualResetEvent evDone = new ManualResetEvent(false);
        Exception exception = null;
        bool      isDone = false;

        #region IOperation Members

        /// <summary>
        /// Determines if the operation has completed.
        /// </summary>
        public bool IsDone
        {
            get { return isDone; }
            protected set { isDone = true; evDone.Set(); }
        }

        /// <summary>
        /// If exception occurs, this will be set.
        /// </summary>
        public Exception ThrownException
        {
            get
            {
                return exception;
            }
            set
            {
                exception = value;
            }
        }

        /// <summary>
        /// Wait till operation has completed.
        /// </summary>
        public bool WaitDone()
        {
            return isDone = evDone.WaitOne();
        }


        /// <summary>
        /// Wait operation to completed with given time-out.
        /// </summary>
        /// <param name="timeOut">time out in milliseconds.</param>
        /// <returns>true if operation completed in given time out, otherwise false.</returns>
        public bool WaitDone(int timeOut)
        {
            return isDone = evDone.WaitOne(timeOut);
        }

        #endregion


        #region IRunnable Members

        /// <summary>
        /// Must implement in derived class.
        /// </summary>
        public abstract void Run();

        void IRunnable.Run()
        {
            try
            {
                Run();
            }
            catch (Exception x)
            {
                this.ThrownException = x;
            }
            finally
            {
                IsDone = true;
            }
        }
   
        #endregion

        /// <summary>
        /// Queues this for execution.
        /// </summary>
        public void Execute()
        {
            SequenceExecutor.Instance.Execute(this);
        }

        /// <summary>
        /// Queues this for execution with given executor.
        /// </summary>
        /// <param name="executer">the executor service to be used.</param>
        public void Execute(IExecutor executer)
        {
            executer.Execute(this);
        }
    
    }
    #endregion

    #region SequenceExecutor, single-threaded per logical cpu, sequential

    /// <summary>
    /// Sequence executor, single-thread per logical cpu, sequential
    /// </summary>
    public class SequenceExecutor : IExecutor
    {
        BlockingQueue<IRunnable> pool;
        List<WorkerThread> threads;

        internal static IExecutor Instance = new SequenceExecutor();

        public static void Execute(IRunnable runnable)
        {
            Instance.Execute(runnable);
        }
            

        public SequenceExecutor()
        {
            pool = new BlockingQueue<IRunnable>();
            threads = new List<WorkerThread>();

            foreach ( var cpu in Range.From(1, QueuedStats.NumberOfProcessors) )
            {
                threads.Add(new WorkerThread(this, cpu));
            }

            threads.ForEach( (thread) => thread.Start() );
        }

        internal IRunnable NextWork()
        {
            return pool.Dequeue();
        }

        #region IExecutorService Members

        void IExecutor.Execute(IRunnable runnable)
        {
            Interlocked.Increment(ref QueuedStats.NumberOfQueuedOperations);

            pool.Enqueue(runnable);


        }

        #endregion

        class WorkerThread : IDisposable
        {
            private SequenceExecutor pool;
            private Thread thread;
            private int[] Cpus;
            private int RunningOnCpu;

            public WorkerThread(SequenceExecutor pool)
            {
                this.Cpus = Range.From(1, QueuedStats.NumberOfProcessors).ToArray();

                this.pool = pool;
                this.thread = new Thread(ThreadMain);
                this.thread.IsBackground = true;
                this.thread.Name = "Executor - Worker Thread";
            }

            public WorkerThread(SequenceExecutor pool, int cpu)
            {
                this.Cpus = new int[cpu];

                this.pool = pool;
                this.thread = new Thread(ThreadMain);
                this.thread.IsBackground = true;
                this.thread.Name = "Executor - Worker Thread";           
            }

            public bool IsAlive
            {
                get { return thread.IsAlive; }
            }

            public string Name
            {
                get
                {
                    return thread.Name + " on CPU:" + RunningOnCpu.ToString();
                }
            }

            public void Start()
            {
                thread.Start();
                Interlocked.Increment(ref QueuedStats.NumberOfRunningThreads);

            }

            public void Stop()
            {
                thread.Abort();
                Interlocked.Decrement(ref QueuedStats.NumberOfRunningThreads);
            }

            void ThreadMain()
            {
                while (true)
                {
                    IRunnable runnable = pool.NextWork();

                    Interlocked.Decrement(ref QueuedStats.NumberOfQueuedOperations);

                    using (ProcessorAffinity.BeginAffinity(Cpus))
                    {
                        this.RunningOnCpu = ProcessorAffinity.GetCurrentProcessorNumber();

                        if (runnable != null)
                            runnable.Run();
                    }

                    Interlocked.Increment(ref QueuedStats.NumberOfCompletedOperations);

                }
            }


            #region IDisposable Members

            public void Dispose()
            {
                Stop();
            }

            #endregion
        }
     
    }
    #endregion

    #region ParallerExecutor, multiple threads per logical cpu, in parallel.
    /// <summary>
    /// Paraller executor, multiple threads per logical cpu, parallel.
    /// </summary>
    public class ParallerExecutor : IExecutor
    {
        //List<WorkerThread> workerThreads = new List<WorkerThread>();

        internal static IExecutor Instance = new ParallerExecutor();

        public static void Execute(IRunnable runnable)
        {
            Instance.Execute(runnable);
        }
            
        #region IExecutorService Members

        void IExecutor.Execute(IRunnable runnable)
        {
            new WorkerThread(runnable).Start();
        }

        #endregion


        class WorkerThread : IDisposable
        {
            private IRunnable runnable;
            private Thread    thread;
            private int[] Cpus;
            private int RunningOnCpu;

            public WorkerThread(IRunnable runnable)
            {
                this.Cpus = Range.From(1, QueuedStats.NumberOfProcessors).ToArray();

                this.runnable = runnable;
                this.thread = new Thread(ThreadMain);
                this.thread.IsBackground = true;
                this.thread.Name = "Executor - Worker Thread";
               
            }
            public bool IsAlive
            {
                get { return thread.IsAlive; }
            }

            public string Name
            {
                get
                {
                    return thread.Name + " on CPU:" + RunningOnCpu.ToString();
                }
            }


            public void Start()
            {
                thread.Start();
                Interlocked.Increment(ref QueuedStats.NumberOfRunningThreads);

            }

            public void Stop()
            {
                thread.Abort();
            }

            void ThreadMain()
            {
                using (ProcessorAffinity.BeginAffinity(Cpus))
                {
                    this.RunningOnCpu = ProcessorAffinity.GetCurrentProcessorNumber();

                    if (runnable != null)
                        runnable.Run();

                }

                Interlocked.Increment(ref QueuedStats.NumberOfCompletedOperations);
                Interlocked.Decrement(ref QueuedStats.NumberOfRunningThreads);

                thread.Join();

            }

            #region IDisposable Members

            public void Dispose()
            {
                Stop();
            }

            #endregion
        }
        
    }
   #endregion

    public class QueuedStats
    {
        public static int NumberOfRunningThreads = 0;
        public static int NumberOfQueuedOperations = 0;
        public static int NumberOfCompletedOperations = 0;
        public static int NumberOfProcessors = Environment.ProcessorCount;
    }

    #region ProcessorAffinity helper classes
    /// <summary>
    /// Gets and sets the processor affinity of the current thread.
    /// </summary>
    public static class ProcessorAffinity
    {
        static class Win32Native
        {
            //GetCurrentThread() returns only a pseudo handle. No need for a SafeHandle here.
            [DllImport("kernel32.dll")]
            public static extern IntPtr GetCurrentThread();

            [DllImport("ntdll.dll", CharSet = CharSet.Auto)]
            public static extern uint NtGetCurrentProcessorNumber();

            [HostProtectionAttribute(SelfAffectingThreading = true)]
            [DllImport("kernel32.dll", CharSet = CharSet.Auto, SetLastError = true)]
            public static extern UIntPtr SetThreadAffinityMask(IntPtr handle, UIntPtr mask);

        }

        public struct ProcessorAffinityHelper : IDisposable
        {
            UIntPtr lastaffinity;

            internal ProcessorAffinityHelper(UIntPtr lastaffinity)
            {
                this.lastaffinity = lastaffinity;
            }

            #region IDisposable Members

            public void Dispose()
            {
                if (lastaffinity != UIntPtr.Zero)
                {
                    Win32Native.SetThreadAffinityMask(Win32Native.GetCurrentThread(), lastaffinity);
                    lastaffinity = UIntPtr.Zero;
                }
            }

            #endregion
        }

        static ulong maskfromids(params int[] ids)
        {
            ulong mask = 0;
            foreach (int id in ids)
            {
                if (id < 0 || id >= Environment.ProcessorCount)
                    throw new ArgumentOutOfRangeException("CPUId", id.ToString());
                mask |= 1UL << id;
            }
            return mask;
        }

        /// <summary>
        /// Sets a processor affinity mask for the current thread.
        /// </summary>
        /// <param name="mask">A thread affinity mask where each bit set to 1 specifies a logical processor on which this thread is allowed to run. 
        /// <remarks>Note: a thread cannot specify a broader set of CPUs than those specified in the process affinity mask.</remarks> 
        /// </param>
        /// <returns>The previous affinity mask for the current thread.</returns>
        public static UIntPtr SetThreadAffinityMask(UIntPtr mask)
        {
            UIntPtr lastaffinity = Win32Native.SetThreadAffinityMask(Win32Native.GetCurrentThread(), mask);
            if (lastaffinity == UIntPtr.Zero)
                throw new Win32Exception(Marshal.GetLastWin32Error());
            return lastaffinity;
        }

        /// <summary>
        /// Sets the logical CPUs that the current thread is allowed to execute on.
        /// </summary>
        /// <param name="CPUIds">One or more logical processor identifier(s) the current thread is allowed to run on.<remarks>Note: numbering starts from 0.</remarks></param>
        /// <returns>The previous affinity mask for the current thread.</returns>
        public static UIntPtr SetThreadAffinity(params int[] CPUIds)
        {
            return SetThreadAffinityMask(((UIntPtr)maskfromids(CPUIds)));
        }

        /// <summary>
        /// Returns the logical CPU number which current thread is running.
        /// </summary>
        public static int GetCurrentProcessorNumber()
        {
            return (int)Win32Native.NtGetCurrentProcessorNumber();
        }

        /// <summary>
        /// Returns the number of logical CPUs on this system.
        /// </summary>
        public static int GetProcessorCount()
        {
            return Environment.ProcessorCount;
        }
        
        /// <summary>
        /// Restrict a code block to run on the specified logical CPUs in conjuction with 
        /// the <code>using</code> statement.
        /// </summary>
        /// <param name="CPUIds">One or more logical processor identifier(s) the current thread is allowed to run on.<remarks>Note: numbering starts from 0.</remarks></param>
        /// <returns>A helper structure that will reset the affinity when its Dispose() method is called at the end of the using block.</returns>
        public static ProcessorAffinityHelper BeginAffinity(params int[] CPUIds)
        {
            return new ProcessorAffinityHelper(SetThreadAffinityMask(((UIntPtr)maskfromids(CPUIds))));
        }

    }

    /// <summary>
    /// Simple range class for integers.
    /// </summary>
    public class Range : IEnumerable<int>
    {
        private int start;
        private int end;
        private int step;
        private bool inclusive;
        
        /// <summary>
        /// Creates new range.
        /// </summary>
        /// <param name="start">the initial value.</param>
        /// <param name="end">the last value.</param>
        /// <param name="step">increase per each step.</param>
        /// <param name="inclusive">include last value in set.</param>
        public Range(int start, int end, int step, bool inclusive = false)
        {
            this.start = start;
            this.end = end;
            this.step = step;
            this.inclusive = inclusive;
        }

        /// <summary>
        /// Creates new range with given values.
        /// </summary>
        /// <param name="start">the start value.</param>
        /// <param name="end">the end value.</param>
        /// <param name="step">the amount to increase per each step.</param>
        /// <param name="includeEnd">should the end value be included in set?</param>
        /// <returns>the range between [start..end]</returns>
        public static IEnumerable<int> From(int start, int end, int step = 1, bool includeEnd = true)
        {
            return new Range(start, end, step, includeEnd);
        }


        /// <summary>
        /// Returns the enumerator for this range.
        /// </summary>
        public IEnumerator<int> GetEnumerator()
        {
            int curr = start;

            do
            {
                yield return curr;

                curr = curr + step;

            } while (curr < end);

            if (inclusive)
                yield return end;

            yield break;
            
        }

        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }
    }
    #endregion
}
